Node / meshtastic / Meshtastic-Android / files / core / network / src / commonTest / kotlin / org / meshtastic / core / network / radio / StreamTransportTest.kt

Displaying Raw • Download

T8b949e/*

* Copyright (c) 2026 Meshtastic LLC

*

* This program is free software: you can redistribute it and/or modify

* it under the terms of the GNU General Public License as published by

* the Free Software Foundation, either version 3 of the License, or

* (at your option) any later version.

*

* This program is distributed in the hope that it will be useful,

* but WITHOUT ANY WARRANTY; without even the implied warranty of

* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

* GNU General Public License for more details.

*

* You should have received a copy of the GNU General Public License

* along with this program. If not, see <https://www.gnu.org/licenses/>.

*/

Tff7b72package T7ee787org.meshtastic.core.network.radio

Tff7b72import T7ee787dev.mokkery.MockMode

Tff7b72import T7ee787dev.mokkery.matcher.any

Tff7b72import T7ee787dev.mokkery.mock

Tff7b72import T7ee787dev.mokkery.verify

Tff7b72import T7ee787dev.mokkery.verify.VerifyMode

Tff7b72import T7ee787io.kotest.property.Arb

Tff7b72import T7ee787io.kotest.property.arbitrary.byte

Tff7b72import T7ee787io.kotest.property.arbitrary.byteArray

Tff7b72import T7ee787io.kotest.property.arbitrary.int

Tff7b72import T7ee787io.kotest.property.checkAll

Tff7b72import T7ee787kotlinx.coroutines.CancellationException

Tff7b72import T7ee787kotlinx.coroutines.CompletableDeferred

Tff7b72import T7ee787kotlinx.coroutines.CoroutineScope

Tff7b72import T7ee787kotlinx.coroutines.async

Tff7b72import T7ee787kotlinx.coroutines.awaitCancellation

Tff7b72import T7ee787kotlinx.coroutines.cancel

Tff7b72import T7ee787kotlinx.coroutines.test.TestScope

Tff7b72import T7ee787kotlinx.coroutines.test.runTest

Tff7b72import T7ee787org.meshtastic.core.network.transport.StreamFrameCodec

Tff7b72import T7ee787org.meshtastic.core.repository.RadioTransportCallback

Tff7b72import T7ee787kotlin.test.Test

Tff7b72import T7ee787kotlin.test.assertEquals

Tff7b72import T7ee787kotlin.test.assertFalse

Tff7b72import T7ee787kotlin.test.assertTrue

Tff7b72class T56d364StreamTransportTest Tb4b4b4{

Tff7b72private Tff7b72val Te6edf3callbackTb4b4b4: Te6edf3RadioTransportCallback Tff7b72= Te6edf3mockTb4b4b4(Te6edf3MockModeTb4b4b4.Te6edf3autofillTb4b4b4)

Tff7b72private Tff7b72lateinit Tff7b72var Te6edf3fakeStreamTb4b4b4: Te6edf3FakeStreamTransport

Tff7b72class T56d364FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4: Te6edf3RadioTransportCallbackTb4b4b4, Te6edf3scopeTb4b4b4: Te6edf3CoroutineScopeTb4b4b4) Tb4b4b4:

Te6edf3StreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3scopeTb4b4b4) Tb4b4b4{

Tff7b72val Te6edf3sentBytes Tff7b72= Te6edf3mutableListOfTff7b72<Te6edf3ByteArrayTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72override Tff7b72fun Td2a8ffsendBytesTb4b4b4(Te6edf3pTb4b4b4: Te6edf3ByteArrayTb4b4b4) Tb4b4b4{

Te6edf3sentBytesTb4b4b4.Te6edf3addTb4b4b4(Te6edf3pTb4b4b4)

Tb4b4b4}

Tff7b72override Tff7b72fun Td2a8ffflushBytesTb4b4b4(Tb4b4b4) Tb4b4b4{

T8b949e/* no-op */

Tb4b4b4}

Tff7b72override Tff7b72fun Td2a8ffkeepAliveTb4b4b4(Tb4b4b4) Tb4b4b4{

T8b949e/* no-op */

Tb4b4b4}

Tff7b72fun Td2a8fffeedTb4b4b4(Te6edf3bTb4b4b4: Tffa657ByteTb4b4b4) Tff7b72= Te6edf3readCharTb4b4b4(Te6edf3bTb4b4b4)

Tff7b72fun Td2a8ffqueueTb4b4b4(Te6edf3payloadTb4b4b4: Te6edf3ByteArrayTb4b4b4, Te6edf3writerTb4b4b4: Te6edf3suspend Tb4b4b4(Te6edf3ByteArrayTb4b4b4) Tff7b72-Tff7b72> Tffa657UnitTb4b4b4, Te6edf3onCompletionTb4b4b4: Tb4b4b4(Tb4b4b4) Tff7b72-Tff7b72> Tffa657UnitTb4b4b4)Tb4b4b4: Tffa657Boolean Tff7b72=

Te6edf3queueFramedSendTb4b4b4(Te6edf3payloadTb4b4b4, Te6edf3writer Tff7b72= Te6edf3writerTb4b4b4, Te6edf3onCompletion Tff7b72= Te6edf3onCompletionTb4b4b4)

Tff7b72public Tff7b72override Tff7b72fun Td2a8ffconnectTb4b4b4(Tb4b4b4) Tff7b72= Tff7b72superTb4b4b4.Te6edf3connectTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`handleSendToRadio property test`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Te6edf3fakeStream Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Te6edf3checkAllTb4b4b4(Te6edf3ArbTb4b4b4.Te6edf3byteArrayTb4b4b4(Te6edf3ArbTb4b4b4.Te6edf3intTb4b4b4(T79c0ff0Tb4b4b4, T79c0ff5T79c0ff1T79c0ff2Tb4b4b4)Tb4b4b4, Te6edf3ArbTb4b4b4.Te6edf3byteTb4b4b4(Tb4b4b4)Tb4b4b4)Tb4b4b4) Tb4b4b4{ Te6edf3payload Tff7b72-Tff7b72>

Te6edf3assertTrueTb4b4b4(Te6edf3fakeStreamTb4b4b4.Te6edf3handleSendToRadioTb4b4b4(Te6edf3payloadTb4b4b4)Tb4b4b4)

Te6edf3testSchedulerTb4b4b4.Te6edf3runCurrentTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`send is rejected after the transport scope stops`Tb4b4b4(Tb4b4b4) Tb4b4b4{

Tff7b72val Te6edf3stoppedScope Tff7b72= Te6edf3TestScopeTb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3transport Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3stoppedScopeTb4b4b4)

Te6edf3stoppedScopeTb4b4b4.Te6edf3cancelTb4b4b4(Tb4b4b4)

Te6edf3assertFalseTb4b4b4(Te6edf3transportTb4b4b4.Te6edf3handleSendToRadioTb4b4b4(Te6edf3byteArrayOfTb4b4b4(T79c0ff1Tb4b4b4, T79c0ff2Tb4b4b4, T79c0ff3Tb4b4b4)Tb4b4b4)Tb4b4b4)

Te6edf3assertTrueTb4b4b4(Te6edf3transportTb4b4b4.Te6edf3sentBytesTb4b4b4.Te6edf3isEmptyTb4b4b4(Tb4b4b4)Tb4b4b4)

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`rejected framed send releases its completion owner exactly once`Tb4b4b4(Tb4b4b4) Tb4b4b4{

Tff7b72val Te6edf3stoppedScope Tff7b72= Te6edf3TestScopeTb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3transport Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3stoppedScopeTb4b4b4)

Tff7b72var Te6edf3completions Tff7b72= T79c0ff0

Te6edf3stoppedScopeTb4b4b4.Te6edf3cancelTb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3accepted Tff7b72= Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(Te6edf3byteArrayOfTb4b4b4(T79c0ff1Tb4b4b4)Tb4b4b4, Te6edf3writer Tff7b72= Tb4b4b4{Tb4b4b4}Tb4b4b4, Te6edf3onCompletion Tff7b72= Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4)

Te6edf3assertFalseTb4b4b4(Te6edf3acceptedTb4b4b4)

Te6edf3assertEqualsTb4b4b4(T79c0ff1Tb4b4b4, Te6edf3completionsTb4b4b4)

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`concurrent framed sends preserve FIFO without interleaving writes`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Tff7b72val Te6edf3transport Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Tff7b72val Te6edf3firstWriteStarted Tff7b72= Te6edf3CompletableDeferredTff7b72<Tffa657UnitTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3releaseFirstWrite Tff7b72= Te6edf3CompletableDeferredTff7b72<Tffa657UnitTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3events Tff7b72= Te6edf3mutableListOfTff7b72<Tffa657StringTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72var Te6edf3firstWrites Tff7b72= T79c0ff0

Tff7b72var Te6edf3completions Tff7b72= T79c0ff0

Te6edf3assertTrueTb4b4b4(

Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(

Te6edf3payload Tff7b72= Te6edf3byteArrayOfTb4b4b4(T79c0ff1Tb4b4b4)Tb4b4b4,

Te6edf3writer Tff7b72= Tb4b4b4{

Te6edf3events Tff7b72+Tff7b72= Ta5d6ff"Ta5d6fffirstTa5d6ff"

Tff7b72if Tb4b4b4(Te6edf3firstWritesTff7b72+Tff7b72+ Tff7b72=Tff7b72= T79c0ff0Tb4b4b4) Tb4b4b4{

Te6edf3firstWriteStartedTb4b4b4.Te6edf3completeTb4b4b4(Tffa657UnitTb4b4b4)

Te6edf3releaseFirstWriteTb4b4b4.Te6edf3awaitTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tb4b4b4}Tb4b4b4,

Te6edf3onCompletion Tff7b72= Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4,

Tb4b4b4)Tb4b4b4,

Tb4b4b4)

Te6edf3assertTrueTb4b4b4(Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(Te6edf3payload Tff7b72= Te6edf3byteArrayOfTb4b4b4(T79c0ff2Tb4b4b4)Tb4b4b4, Te6edf3writer Tff7b72= Tb4b4b4{ Te6edf3events Tff7b72+Tff7b72= Ta5d6ff"Ta5d6ffsecondTa5d6ff" Tb4b4b4}Tb4b4b4) Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4)

Te6edf3firstWriteStartedTb4b4b4.Te6edf3awaitTb4b4b4(Tb4b4b4)

Te6edf3assertEqualsTb4b4b4(Te6edf3listOfTb4b4b4(Ta5d6ff"Ta5d6fffirstTa5d6ff"Tb4b4b4)Tb4b4b4, Te6edf3eventsTb4b4b4)

Te6edf3releaseFirstWriteTb4b4b4.Te6edf3completeTb4b4b4(Tffa657UnitTb4b4b4)

Te6edf3testSchedulerTb4b4b4.Te6edf3runCurrentTb4b4b4(Tb4b4b4)

T8b949e// StreamFrameCodec invokes the writer once for the frame header and once for the body of each payload.

Te6edf3assertEqualsTb4b4b4(Te6edf3listOfTb4b4b4(Ta5d6ff"Ta5d6fffirstTa5d6ff"Tb4b4b4, Ta5d6ff"Ta5d6fffirstTa5d6ff"Tb4b4b4, Ta5d6ff"Ta5d6ffsecondTa5d6ff"Tb4b4b4, Ta5d6ff"Ta5d6ffsecondTa5d6ff"Tb4b4b4)Tb4b4b4, Te6edf3eventsTb4b4b4)

Te6edf3assertEqualsTb4b4b4(T79c0ff2Tb4b4b4, Te6edf3completionsTb4b4b4)

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`framed send queue reports backpressure at its capacity`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Tff7b72val Te6edf3transport Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Tff7b72val Te6edf3firstWriteStarted Tff7b72= Te6edf3CompletableDeferredTff7b72<Tffa657UnitTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3releaseFirstWrite Tff7b72= Te6edf3CompletableDeferredTff7b72<Tffa657UnitTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72var Te6edf3acceptedCompletions Tff7b72= T79c0ff0

Tff7b72var Te6edf3rejectedCompletions Tff7b72= T79c0ff0

Te6edf3assertTrueTb4b4b4(

Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(

Te6edf3payload Tff7b72= Te6edf3byteArrayOfTb4b4b4(T79c0ff0Tb4b4b4)Tb4b4b4,

Te6edf3writer Tff7b72= Tb4b4b4{

Te6edf3firstWriteStartedTb4b4b4.Te6edf3completeTb4b4b4(Tffa657UnitTb4b4b4)

Te6edf3releaseFirstWriteTb4b4b4.Te6edf3awaitTb4b4b4(Tb4b4b4)

Tb4b4b4}Tb4b4b4,

Te6edf3onCompletion Tff7b72= Tb4b4b4{Tb4b4b4}Tb4b4b4,

Tb4b4b4)Tb4b4b4,

Tb4b4b4)

Te6edf3firstWriteStartedTb4b4b4.Te6edf3awaitTb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3completionsByIndex Tff7b72= Te6edf3IntArrayTb4b4b4(Te6edf3StreamTransportTb4b4b4.Te6edf3MAX_PENDING_SENDS Tff7b72+ T79c0ff1Tb4b4b4)

T8b949e// The worker has dequeued the blocked first send, so the bounded channel has exactly MAX_PENDING_SENDS slots.

Tff7b72val Te6edf3queuedAdmissions Tff7b72=

Te6edf3ListTb4b4b4(Te6edf3StreamTransportTb4b4b4.Te6edf3MAX_PENDING_SENDS Tff7b72+ T79c0ff1Tb4b4b4) Tb4b4b4{ Te6edf3index Tff7b72-Tff7b72>

Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(Te6edf3byteArrayOfTb4b4b4(Te6edf3indexTb4b4b4.Te6edf3toByteTb4b4b4(Tb4b4b4)Tb4b4b4)Tb4b4b4, Te6edf3writer Tff7b72= Tb4b4b4{Tb4b4b4}Tb4b4b4) Tb4b4b4{ Te6edf3completionsByIndexTff7b72[Te6edf3indexTff7b72]Tff7b72+Tff7b72+ Tb4b4b4}

Tb4b4b4}

Te6edf3queuedAdmissionsTb4b4b4.Te6edf3forEachIndexed Tb4b4b4{ Te6edf3indexTb4b4b4, Te6edf3accepted Tff7b72-Tff7b72>

Tff7b72if Tb4b4b4(Te6edf3acceptedTb4b4b4) Tb4b4b4{

Te6edf3acceptedCompletions Tff7b72+Tff7b72= Te6edf3completionsByIndexTff7b72[Te6edf3indexTff7b72]

Tb4b4b4} Tff7b72else Tb4b4b4{

Te6edf3rejectedCompletions Tff7b72+Tff7b72= Te6edf3completionsByIndexTff7b72[Te6edf3indexTff7b72]

Tb4b4b4}

Tb4b4b4}

Te6edf3assertEqualsTb4b4b4(Te6edf3StreamTransportTb4b4b4.Te6edf3MAX_PENDING_SENDSTb4b4b4, Te6edf3queuedAdmissionsTb4b4b4.Te6edf3count Tb4b4b4{ Tffa657it Tb4b4b4}Tb4b4b4)

Te6edf3assertFalseTb4b4b4(Te6edf3queuedAdmissionsTb4b4b4.Te6edf3lastTb4b4b4(Tb4b4b4)Tb4b4b4, Ta5d6ff"Ta5d6ffthe first send beyond capacity must be rejectedTa5d6ff"Tb4b4b4)

Te6edf3assertEqualsTb4b4b4(T79c0ff1Tb4b4b4, Te6edf3rejectedCompletionsTb4b4b4, Ta5d6ff"Ta5d6ffa capacity-rejected send must release its owner exactly onceTa5d6ff"Tb4b4b4)

Te6edf3assertEqualsTb4b4b4(T79c0ff0Tb4b4b4, Te6edf3acceptedCompletionsTb4b4b4, Ta5d6ff"Ta5d6ffadmitted sends must stay queued while the worker is blockedTa5d6ff"Tb4b4b4)

Te6edf3releaseFirstWriteTb4b4b4.Te6edf3completeTb4b4b4(Tffa657UnitTb4b4b4)

Te6edf3testSchedulerTb4b4b4.Te6edf3runCurrentTb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3admittedCompletions Tff7b72=

Te6edf3queuedAdmissions

Tb4b4b4.Te6edf3withIndexTb4b4b4(Tb4b4b4)

Tb4b4b4.Te6edf3filter Tb4b4b4{ Tb4b4b4(Te6edf3_Tb4b4b4, Te6edf3acceptedTb4b4b4) Tff7b72-Tff7b72> Te6edf3accepted Tb4b4b4}

Tb4b4b4.Te6edf3sumOf Tb4b4b4{ Tb4b4b4(Te6edf3indexTb4b4b4, Te6edf3_Tb4b4b4) Tff7b72-Tff7b72> Te6edf3completionsByIndexTff7b72[Te6edf3indexTff7b72] Tb4b4b4}

Te6edf3assertEqualsTb4b4b4(Te6edf3StreamTransportTb4b4b4.Te6edf3MAX_PENDING_SENDSTb4b4b4, Te6edf3admittedCompletionsTb4b4b4)

Te6edf3assertEqualsTb4b4b4(T79c0ff1Tb4b4b4, Te6edf3completionsByIndexTb4b4b4.Te6edf3lastTb4b4b4(Tb4b4b4)Tb4b4b4, Ta5d6ff"Ta5d6ffthe rejected send must release its owner exactly onceTa5d6ff"Tb4b4b4)

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`close lets an active framed send finish before returning`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Tff7b72val Te6edf3transport Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Tff7b72val Te6edf3writeStarted Tff7b72= Te6edf3CompletableDeferredTff7b72<Tffa657UnitTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3releaseWrite Tff7b72= Te6edf3CompletableDeferredTff7b72<Tffa657UnitTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72var Te6edf3completions Tff7b72= T79c0ff0

Te6edf3assertTrueTb4b4b4(

Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(

Te6edf3payload Tff7b72= Te6edf3byteArrayOfTb4b4b4(T79c0ff1Tb4b4b4)Tb4b4b4,

Te6edf3writer Tff7b72= Tb4b4b4{

Te6edf3writeStartedTb4b4b4.Te6edf3completeTb4b4b4(Tffa657UnitTb4b4b4)

Te6edf3releaseWriteTb4b4b4.Te6edf3awaitTb4b4b4(Tb4b4b4)

Tb4b4b4}Tb4b4b4,

Te6edf3onCompletion Tff7b72= Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4,

Tb4b4b4)Tb4b4b4,

Tb4b4b4)

Te6edf3writeStartedTb4b4b4.Te6edf3awaitTb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3closeJob Tff7b72= Te6edf3async Tb4b4b4{ Te6edf3transportTb4b4b4.Te6edf3closeTb4b4b4(Tb4b4b4) Tb4b4b4}

Te6edf3testSchedulerTb4b4b4.Te6edf3runCurrentTb4b4b4(Tb4b4b4)

Te6edf3assertFalseTb4b4b4(Te6edf3closeJobTb4b4b4.Te6edf3isCompletedTb4b4b4, Ta5d6ff"Ta5d6ffclose must not cancel an active frame while it can still drainTa5d6ff"Tb4b4b4)

Te6edf3releaseWriteTb4b4b4.Te6edf3completeTb4b4b4(Tffa657UnitTb4b4b4)

Te6edf3closeJobTb4b4b4.Te6edf3awaitTb4b4b4(Tb4b4b4)

Te6edf3assertEqualsTb4b4b4(T79c0ff1Tb4b4b4, Te6edf3completionsTb4b4b4)

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`close completes ownership when the worker never started`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Tff7b72val Te6edf3transport Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Tff7b72var Te6edf3completions Tff7b72= T79c0ff0

Te6edf3assertTrueTb4b4b4(Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(Te6edf3byteArrayOfTb4b4b4(T79c0ff1Tb4b4b4)Tb4b4b4, Te6edf3writer Tff7b72= Tb4b4b4{Tb4b4b4}Tb4b4b4, Te6edf3onCompletion Tff7b72= Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4)Tb4b4b4)

Te6edf3assertTrueTb4b4b4(Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(Te6edf3byteArrayOfTb4b4b4(T79c0ff2Tb4b4b4)Tb4b4b4, Te6edf3writer Tff7b72= Tb4b4b4{Tb4b4b4}Tb4b4b4, Te6edf3onCompletion Tff7b72= Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4)Tb4b4b4)

Te6edf3transportTb4b4b4.Te6edf3closeTb4b4b4(Tb4b4b4)

Te6edf3assertEqualsTb4b4b4(T79c0ff2Tb4b4b4, Te6edf3completionsTb4b4b4, Ta5d6ff"Ta5d6ffshutdown must release owners even before the worker startsTa5d6ff"Tb4b4b4)

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`per-send cancellation does not stop the framed-send worker`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Tff7b72val Te6edf3transport Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Tff7b72var Te6edf3completions Tff7b72= T79c0ff0

Tff7b72val Te6edf3laterWrites Tff7b72= Te6edf3mutableListOfTff7b72<Te6edf3ByteArrayTff7b72>Tb4b4b4(Tb4b4b4)

Te6edf3assertTrueTb4b4b4(

Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(

Te6edf3payload Tff7b72= Te6edf3byteArrayOfTb4b4b4(T79c0ff1Tb4b4b4)Tb4b4b4,

Te6edf3writer Tff7b72= Tb4b4b4{ Tff7b72throw Te6edf3CancellationExceptionTb4b4b4(Ta5d6ff"Ta5d6ffcancel this send onlyTa5d6ff"Tb4b4b4) Tb4b4b4}Tb4b4b4,

Te6edf3onCompletion Tff7b72= Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4,

Tb4b4b4)Tb4b4b4,

Tb4b4b4)

Te6edf3assertTrueTb4b4b4(

Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(Te6edf3payload Tff7b72= Te6edf3byteArrayOfTb4b4b4(T79c0ff2Tb4b4b4)Tb4b4b4, Te6edf3writer Tff7b72= Tb4b4b4{ Te6edf3laterWrites Tff7b72+Tff7b72= Tffa657it Tb4b4b4}Tb4b4b4, Te6edf3onCompletion Tff7b72= Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4)Tb4b4b4,

Tb4b4b4)

Te6edf3testSchedulerTb4b4b4.Te6edf3runCurrentTb4b4b4(Tb4b4b4)

Te6edf3assertTrueTb4b4b4(Te6edf3laterWritesTb4b4b4.Te6edf3isNotEmptyTb4b4b4(Tb4b4b4)Tb4b4b4, Ta5d6ff"Ta5d6ffa cancelled send must not permanently stop the workerTa5d6ff"Tb4b4b4)

Te6edf3assertEqualsTb4b4b4(T79c0ff2Tb4b4b4, Te6edf3completionsTb4b4b4)

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`readChar property test`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Te6edf3fakeStream Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Te6edf3checkAllTb4b4b4(Te6edf3ArbTb4b4b4.Te6edf3byteArrayTb4b4b4(Te6edf3ArbTb4b4b4.Te6edf3intTb4b4b4(T79c0ff0Tb4b4b4, T79c0ff1T79c0ff0T79c0ff0Tb4b4b4)Tb4b4b4, Te6edf3ArbTb4b4b4.Te6edf3byteTb4b4b4(Tb4b4b4)Tb4b4b4)Tb4b4b4) Tb4b4b4{ Te6edf3data Tff7b72-Tff7b72>

Te6edf3dataTb4b4b4.Te6edf3forEach Tb4b4b4{ Te6edf3fakeStreamTb4b4b4.Te6edf3feedTb4b4b4(Tffa657itTb4b4b4) Tb4b4b4}

T8b949e// Ensure no crash

Tb4b4b4}

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`connect sends wake bytes`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Te6edf3fakeStream Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Te6edf3fakeStreamTb4b4b4.Te6edf3connectTb4b4b4(Tb4b4b4)

Te6edf3assertTrueTb4b4b4(Te6edf3fakeStreamTb4b4b4.Te6edf3sentBytesTb4b4b4.Te6edf3isNotEmptyTb4b4b4(Tb4b4b4)Tb4b4b4)

Te6edf3assertTrueTb4b4b4(Te6edf3fakeStreamTb4b4b4.Te6edf3sentBytesTff7b72[T79c0ff0Tff7b72]Tb4b4b4.Te6edf3contentEqualsTb4b4b4(Te6edf3StreamFrameCodecTb4b4b4.Te6edf3WAKE_BYTESTb4b4b4)Tb4b4b4)

Te6edf3verify Tb4b4b4{ Te6edf3callbackTb4b4b4.Te6edf3onConnectTb4b4b4(Tb4b4b4) Tb4b4b4}

Te6edf3fakeStreamTb4b4b4.Te6edf3closeTb4b4b4(Tb4b4b4)

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`close does not emit disconnect callback`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Te6edf3fakeStream Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Te6edf3fakeStreamTb4b4b4.Te6edf3connectTb4b4b4(Tb4b4b4)

Te6edf3fakeStreamTb4b4b4.Te6edf3closeTb4b4b4(Tb4b4b4)

Te6edf3verifyTb4b4b4(Te6edf3mode Tff7b72= Te6edf3VerifyModeTb4b4b4.Te6edf3notTb4b4b4) Tb4b4b4{

Te6edf3callbackTb4b4b4.Te6edf3onDisconnectTb4b4b4(Te6edf3isPermanent Tff7b72= Te6edf3anyTb4b4b4(Tb4b4b4)Tb4b4b4, Te6edf3errorMessage Tff7b72= Te6edf3anyTb4b4b4(Tb4b4b4)Tb4b4b4, Te6edf3reason Tff7b72= Te6edf3anyTb4b4b4(Tb4b4b4)Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

Tf0883e@Test

Tff7b72fun Td2a8ff`close completes ownership for in-flight and queued framed sends`Tb4b4b4(Tb4b4b4) Tff7b72= Te6edf3runTest Tb4b4b4{

Tff7b72val Te6edf3transport Tff7b72= Te6edf3FakeStreamTransportTb4b4b4(Te6edf3callbackTb4b4b4, Te6edf3backgroundScopeTb4b4b4)

Tff7b72val Te6edf3firstWriteStarted Tff7b72= Te6edf3CompletableDeferredTff7b72<Tffa657UnitTff7b72>Tb4b4b4(Tb4b4b4)

Tff7b72var Te6edf3completions Tff7b72= T79c0ff0

Te6edf3assertTrueTb4b4b4(

Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(

Te6edf3payload Tff7b72= Te6edf3byteArrayOfTb4b4b4(T79c0ff1Tb4b4b4)Tb4b4b4,

Te6edf3writer Tff7b72= Tb4b4b4{

Te6edf3firstWriteStartedTb4b4b4.Te6edf3completeTb4b4b4(Tffa657UnitTb4b4b4)

Te6edf3awaitCancellationTb4b4b4(Tb4b4b4)

Tb4b4b4}Tb4b4b4,

Te6edf3onCompletion Tff7b72= Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4,

Tb4b4b4)Tb4b4b4,

Tb4b4b4)

Te6edf3assertTrueTb4b4b4(Te6edf3transportTb4b4b4.Te6edf3queueTb4b4b4(Te6edf3byteArrayOfTb4b4b4(T79c0ff2Tb4b4b4)Tb4b4b4, Te6edf3writer Tff7b72= Tb4b4b4{Tb4b4b4}Tb4b4b4, Te6edf3onCompletion Tff7b72= Tb4b4b4{ Te6edf3completionsTff7b72+Tff7b72+ Tb4b4b4}Tb4b4b4)Tb4b4b4)

Te6edf3firstWriteStartedTb4b4b4.Te6edf3awaitTb4b4b4(Tb4b4b4)

Te6edf3transportTb4b4b4.Te6edf3closeTb4b4b4(Tb4b4b4)

Te6edf3assertEqualsTb4b4b4(T79c0ff2Tb4b4b4, Te6edf3completionsTb4b4b4, Ta5d6ff"Ta5d6ffshutdown must release every accepted framed-send owner exactly onceTa5d6ff"Tb4b4b4)

Tb4b4b4}

Tb4b4b4}